/*-------------------------------------------------------------------------
 * tablesync.c
 *	  PostgreSQL逻辑复制：初始表数据同步
 *
 * Copyright (c) 2012-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/logical/tablesync.c
 *
 * NOTES
 *	  此文件包含用于逻辑复制的初始表数据同步的代码。
 *
 *	  初始数据同步是为每个表单独完成的，
 *	  在一个单独的应用工作进程中，仅从发布者提取初始
 *	  快照数据，然后将流中的位置与主应用工作进程进行同步。
 *
 *	  以这种方式进行同步有几个原因：
 *	   - 允许我们并行化初始数据同步，降低了完成的时间。
 *	   - 初始同步不必在复制所有表数据所需的时间内保持xid和LSN，
 *		 造成更少的膨胀和较低的磁盘消耗，相比之下，在整个数据库中
 *		 进行整个过程的同步。
 *	   - 允许我们在初始同步完成后同步任何添加的表。
 *
 *	  流位置同步分多个步骤进行：
 *	   - 应用工作进程请求表同步工作进程开始，
 *		 将新表状态设置为INIT。
 *	   - 表同步工作进程启动；在复制过程中将表状态从INIT更改
 *		 为DATASYNC。
 *	   - 表同步工作进程进行初始表复制；有一个FINISHEDCOPY
 *		 （特定于同步工作）的状态，指示复制阶段已完成，
 *		 所以如果工作进程以这个（非内存）状态崩溃，
 *		 则复制将不会被重新尝试。
 *	   - 表同步工作进程然后将表状态设置为SYNCWAIT；等待状态更改。
 *	   - 应用工作进程定期检查处于SYNCWAIT状态的表。
 *		 当有表出现时，它将表状态设置为CATCHUP并开始循环等待，
 *		 直到表状态设置为SYNCDONE或同步工作进程退出。
 *	   - 一旦同步工作进程看到状态变化为CATCHUP，
 *		 它将读取流并应用更改（像应用工作进程一样），
 *		 直到跟上指定的流位置。然后将状态设置为SYNCDONE。
 *		 在CATCHUP和SYNCDONE之间可能没有应用更改，
 *		 因为同步工作进程可能在应用工作进程之前。
 *	   - 一旦状态设置为SYNCDONE，应用将继续跟踪表，
 *		 直到达到SYNCDONE流位置，此时将状态设置为READY并停止跟踪。
 *		 当然，在此过程中可能没有变化。
 *
 *	  所以状态进展始终是：INIT -> DATASYNC -> FINISHEDCOPY
 *	  -> SYNCWAIT -> CATCHUP -> SYNCDONE -> READY。
 *
 *	  catalog pg_subscription_rel 用于保持关于
 *	  订阅表及其状态的信息。该目录持有所有状态，
 *	  除了SYNCWAIT和CATCHUP，它们仅在共享内存中。
 *
 *	  示例流程如下：
 *	   - 应用在前：
 *		  sync:8
 *			-> 在目录中设置为FINISHEDCOPY
 *			-> 在内存中设置为SYNCWAIT
 *		  apply:10
 *			-> 在内存中设置为CATCHUP
 *			-> 进入等待循环
 *		  sync:10
 *			-> 在目录中设置为SYNCDONE
 *			-> 退出
 *		  apply:10
 *			-> 退出等待循环
 *			-> 继续复制
 *		  apply:11
 *			-> 在目录中设置为READY
 *
 *	   - 同步在前：
 *		  sync:10
 *			-> 在目录中设置为FINISHEDCOPY
 *			-> 在内存中设置为SYNCWAIT
 *		  apply:8
 *			-> 在内存中设置为CATCHUP
 *			-> 继续按表过滤
 *		  sync:10
 *			-> 在目录中设置为SYNCDONE
 *			-> 退出
 *		  apply:10
 *			-> 在目录中设置为READY
 *			-> 停止按表过滤
 *			-> 继续复制
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/table.h"
#include "access/xact.h"
#include "catalog/indexing.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
#include "commands/copy.h"
#include "miscadmin.h"
#include "parser/parse_relation.h"
#include "pgstat.h"
#include "replication/logicallauncher.h"
#include "replication/logicalrelation.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
#include "replication/slot.h"
#include "replication/origin.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "utils/acl.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rls.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"

typedef enum
{
	SYNC_TABLE_STATE_NEEDS_REBUILD,
	SYNC_TABLE_STATE_REBUILD_STARTED,
	SYNC_TABLE_STATE_VALID,
} SyncingTablesState;

static SyncingTablesState table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
static List *table_states_not_ready = NIL;
static bool fc_FetchTableStates(bool *fc_started_tx);

static StringInfo copybuf = NULL;

/*
 * 同步工作者的退出例程。
 */
static void pg_attribute_noreturn()
fc_finish_sync_worker(void)
{
	/*
	 * 提交任何未完成的事务。这是通常情况，除非
	 * 表没有任何操作。
	 */
	if (IsTransactionState())
	{
		CommitTransactionCommand();
		pgstat_report_stat(true);
	}

	/* 并刷新所有写入。 */
	XLogFlush(GetXLogWriteRecPtr());

	StartTransactionCommand();
	ereport(LOG,
			(errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished",
					MySubscription->name,
					get_rel_name(MyLogicalRepWorker->relid))));
	CommitTransactionCommand();

	/* 找到主应用工作者并向其发送信号。 */
	logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);

	/* 优雅地停止 */
	proc_exit(0);
}

/*
 * 等待直到关系同步状态在目录中设置为预期的
 * 状态；发生时返回真。
 *
 * 如果表同步工作者或表本身已经消失，或表状态已被重置，
 * 则返回假。
 *
 * 目前，这将在应用工作者中使用，当从
 * CATCHUP 状态转换到 SYNCDONE。
 */
static bool fc_wait_for_relation_state_change(Oid fc_relid, char fc_expected_state)
{
	char		fc_state;

	for (;;)
	{
		LogicalRepWorker *fc_worker;
		XLogRecPtr	fc_statelsn;

		CHECK_FOR_INTERRUPTS();

		InvalidateCatalogSnapshot();
		fc_state = GetSubscriptionRelState(MyLogicalRepWorker->subid,
										fc_relid, &fc_statelsn);

		if (fc_state == SUBREL_STATE_UNKNOWN)
			break;

		if (fc_state == fc_expected_state)
			return true;

		/* 检查同步工作者是否仍在运行，如果没有则退出。 */
		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
		fc_worker = logicalrep_worker_find(MyLogicalRepWorker->subid, fc_relid,
										false);
		LWLockRelease(LogicalRepWorkerLock);
		if (!fc_worker)
			break;

		(void) WaitLatch(MyLatch,
						 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
						 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);

		ResetLatch(MyLatch);
	}

	return false;
}

/*
 * 等待直到应用工作者将我们的同步
 * 工作者的状态更改为预期的状态。
 *
 * 在从 SYNCWAIT 状态转换到 CATCHUP 时使用。
 *
 * 如果应用工作者已经消失，则返回假。
 */
static bool fc_wait_for_worker_state_change(char fc_expected_state)
{
	int			fc_rc;

	for (;;)
	{
		LogicalRepWorker *fc_worker;

		CHECK_FOR_INTERRUPTS();

		/*
		 * 如果已经在正确状态，则完成。 （我们假设这个获取是原子的，
		 * 足够确保如果没有锁定我们不会得到误导性的答案。）
		 */
		if (MyLogicalRepWorker->relstate == fc_expected_state)
			return true;

		/*
		 * 如果应用工作者已经死亡则退出，否则向其发送信号我们
		 * 正在等待。
		 */
		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
		fc_worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
										InvalidOid, false);
		if (fc_worker && fc_worker->proc)
			logicalrep_worker_wakeup_ptr(fc_worker);
		LWLockRelease(LogicalRepWorkerLock);
		if (!fc_worker)
			break;

		/*
		 * 等待。我们期望从应用工作者那里收到一个闩锁信号返回，
		 * 但如果它在没有发送信号的情况下死亡，则使用超时。
		 */
		fc_rc = WaitLatch(MyLatch,
					   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
					   1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);

		if (fc_rc & WL_LATCH_SET)
			ResetLatch(MyLatch);
	}

	return false;
}

/*
 * 来自系统缓存失效的回调。
 */
void invalidate_syncing_table_states(Datum fc_arg, int fc_cacheid, uint32 fc_hashvalue)
{
	table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
}

/*
 * 处理来自同步
 * 工作者的表同步协作。
 *
 * 如果同步工作者处于 CATCHUP 状态并达到了（或超过了）
 * WAL 流中的预定同步点，则将表标记为
 * SYNCDONE 并完成。
 */
static void fc_process_syncing_tables_for_sync(XLogRecPtr fc_current_lsn)
{
	SpinLockAcquire(&MyLogicalRepWorker->relmutex);

	if (MyLogicalRepWorker->relstate == SUBREL_STATE_CATCHUP &&
		fc_current_lsn >= MyLogicalRepWorker->relstate_lsn)
	{
		TimeLineID	fc_tli;
		char		fc_syncslotname[NAMEDATALEN] = {0};

		MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCDONE;
		MyLogicalRepWorker->relstate_lsn = fc_current_lsn;

		SpinLockRelease(&MyLogicalRepWorker->relmutex);

		/*
		 * UpdateSubscriptionRelState 必须在事务内调用。
		 * 此事务将在 finish_sync_worker() 中结束。
		 */
		if (!IsTransactionState())
			StartTransactionCommand();

		UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
								   MyLogicalRepWorker->relid,
								   MyLogicalRepWorker->relstate,
								   MyLogicalRepWorker->relstate_lsn);

		/*
		 * 结束流以便可以使用 LogRepWorkerWalRcvConn 来丢弃
		 * 槽。
		 */
		walrcv_endstreaming(LogRepWorkerWalRcvConn, &fc_tli);

		/*
		 * 清理表同步槽。
		 *
		 * 这必须在更新状态后完成，因为否则如果
		 * 在进行数据库操作时出现错误，我们将无法
		 * 回滚丢弃的槽。
		 */
		ReplicationSlotNameForTablesync(MyLogicalRepWorker->subid,
										MyLogicalRepWorker->relid,
										fc_syncslotname,
										sizeof(fc_syncslotname));

		/*
		 * 如果我们无法丢弃槽，则给出错误是重要的，
		 * 否则，它将不会被丢弃，直到相应的订阅
		 * 被丢弃。所以将 missing_ok 设为假。
		 */
		ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, fc_syncslotname, false);

		fc_finish_sync_worker();
	}
	else
		SpinLockRelease(&MyLogicalRepWorker->relmutex);
}

/*
 * 处理来自应用工作线程的表同步协作。
 *
 * 遍历所有由应用进程单独跟踪的订阅表（目前，所有状态不是
 * SUBREL_STATE_READY 的表）并管理它们的同步。
 *
 * 如果有需要同步的表尚未被同步，启动它们的同步工作线程（如果有空闲的同步
 * 工作线程）。为了防止在故障后以高频率启动相同关系的同步工作线程，
 * 我们将其最后启动时间存储在每个同步状态信息中。我们在等待至少 wal_retrieve_retry_interval 后
 * 启动相同关系的同步工作线程。
 *
 * 对于已经在同步的表，检查同步工作线程是否需要应用工作线程的操作或已完成。
 * 这是 SYNCWAIT 到 CATCHUP 的过渡。
 *
 * 如果达到同步位置（SYNCDONE），那么可以将表标记为 READY，不再进行跟踪。
 */
static void fc_process_syncing_tables_for_apply(XLogRecPtr fc_current_lsn)
{
	struct tablesync_start_time_mapping
	{
		Oid			relid;
		TimestampTz last_start_time;
	};
	static HTAB *fc_last_start_times = NULL;
	ListCell   *fc_lc;
	bool		fc_started_tx = false;

	Assert(!IsTransactionState());

	/* 我们在这里需要订阅表的最新同步状态信息。 */
	fc_FetchTableStates(&fc_started_tx);

	/*
	 * 为跟踪工作线程的最后启动时间准备一个哈希表，以避免
	 * 立即重启。如果没有需要同步的表，我们就不需要它。
	 */
	if (table_states_not_ready && !fc_last_start_times)
	{
		HASHCTL		fc_ctl;

		fc_ctl.keysize = sizeof(Oid);
		fc_ctl.entrysize = sizeof(struct tablesync_start_time_mapping);
		fc_last_start_times = hash_create("Logical replication table sync worker start times",
									   256, &fc_ctl, HASH_ELEM | HASH_BLOBS);
	}

	/*
	 * 当我们完成所有表的处理时清理哈希表（只是为了
	 * 释放一点内存）。
	 */
	else if (!table_states_not_ready && fc_last_start_times)
	{
		hash_destroy(fc_last_start_times);
		fc_last_start_times = NULL;
	}

	/*
	 * 即使用户请求了 two_phase 模式，它在所有表同步达到 READY 状态之前仍保持为
	 * 'pending'。
	 *
	 * 当这种情况发生时，我们重新启动应用工作线程，并且（如果条件
	 * 仍然合适），那么那个时候 two_phase 三态将变为 'enabled'。
	 *
	 * 注意：如果订阅没有表，则将状态保持为
	 * PENDING，这样 ALTER SUBSCRIPTION ... REFRESH PUBLICATION 就可以
	 * 正常工作。
	 */
	if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING &&
		AllTablesyncsReady())
	{
		ereport(LOG,
				(errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled",
						MySubscription->name)));

		proc_exit(0);
	}

	/*
	 * 处理所有正在同步的表。
	 */
	foreach(fc_lc, table_states_not_ready)
	{
		SubscriptionRelState *fc_rstate = (SubscriptionRelState *) lfirst(fc_lc);

		if (fc_rstate->state == SUBREL_STATE_SYNCDONE)
		{
			/*
			 * 应用已跟进到表同步完成的位置。
			 * 将表标记为准备好，以便应用将继续正常地复制它。
			 */
			if (fc_current_lsn >= fc_rstate->lsn)
			{
				char		fc_originname[NAMEDATALEN];

				fc_rstate->state = SUBREL_STATE_READY;
				fc_rstate->lsn = fc_current_lsn;
				if (!fc_started_tx)
				{
					StartTransactionCommand();
					fc_started_tx = true;
				}

				/*
				 * 如果存在，则移除表同步的来源跟踪。
				 *
				 * 正常情况下，来源的移除在这里完成，而不是在
				 * process_syncing_tables_for_sync 函数中，因为我们不
				 * 允许在拥有来源的进程存活之前移除来源。
				 *
				 * 可能用户正在并发执行
				 * 刷新操作来处理我们移除表状态和其来源的订阅，这时来源可能
				 * 已经被移除。因此，传递 missing_ok = true。
				 */
				ReplicationOriginNameForTablesync(MyLogicalRepWorker->subid,
												  fc_rstate->relid,
												  fc_originname,
												  sizeof(fc_originname));
				replorigin_drop_by_name(fc_originname, true, false);

				/*
				 * 仅在源清理之后将状态更新为 READY。
				 */
				UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
										   fc_rstate->relid, fc_rstate->state,
										   fc_rstate->lsn);
			}
		}
		else
		{
			LogicalRepWorker *fc_syncworker;

			/*
			 * 查找该关系的同步工作线程。
			 */
			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);

			fc_syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid,
												fc_rstate->relid, false);

			if (fc_syncworker)
			{
				/* 找到一个，更新我们对其状态的副本 */
				SpinLockAcquire(&fc_syncworker->relmutex);
				fc_rstate->state = fc_syncworker->relstate;
				fc_rstate->lsn = fc_syncworker->relstate_lsn;
				if (fc_rstate->state == SUBREL_STATE_SYNCWAIT)
				{
					/*
					 * 同步工作线程在等待应用。告诉同步工作线程
					 * 现在可以赶上了。
					 */
					fc_syncworker->relstate = SUBREL_STATE_CATCHUP;
					fc_syncworker->relstate_lsn =
						Max(fc_syncworker->relstate_lsn, fc_current_lsn);
				}
				SpinLockRelease(&fc_syncworker->relmutex);

				/* 如果我们告诉工作线程赶上，等待它。 */
				if (fc_rstate->state == SUBREL_STATE_SYNCWAIT)
				{
					/* 通知同步工作线程，因为它可能在等待我们。 */
					if (fc_syncworker->proc)
						logicalrep_worker_wakeup_ptr(fc_syncworker);

					/* 现在可以安全地释放LWLock */
					LWLockRelease(LogicalRepWorkerLock);

					if (fc_started_tx)
					{
						/*
						 * 我们必须提交现有事务以释放
						 * 当前的锁，在进入繁忙循环之前。
						 * 这是为了避免由于任何现有锁导致的
						 * 未检测到的死锁，因为死锁检测器无法
						 * 检测在闩上的等待。
						 */
						CommitTransactionCommand();
						pgstat_report_stat(false);
					}

					/*
					 * 进入繁忙循环，等待同步工作线程
					 * 达到预期状态（或死去的尝试）。
					 */
					StartTransactionCommand();
					fc_started_tx = true;

					fc_wait_for_relation_state_change(fc_rstate->relid,
												   SUBREL_STATE_SYNCDONE);
				}
				else
					LWLockRelease(LogicalRepWorkerLock);
			}
			else
			{
				/*
				 * 如果此表尚无同步工作线程，请在
				 * 我们拥有锁的同时，计算该订阅的
				 * 正在运行的同步工作线程数量。
				 */
				int			fc_nsyncworkers =
				logicalrep_sync_worker_count(MyLogicalRepWorker->subid);

				/* 现在可以安全地释放LWLock */
				LWLockRelease(LogicalRepWorkerLock);

				/*
				 * 如果有空闲的同步工作线程插槽，
				 * 则为该表启动一个新的同步工作线程。
				 */
				if (fc_nsyncworkers < max_sync_workers_per_subscription)
				{
					TimestampTz fc_now = GetCurrentTimestamp();
					struct tablesync_start_time_mapping *fc_hentry;
					bool		fc_found;

					fc_hentry = hash_search(fc_last_start_times, &fc_rstate->relid,
										 HASH_ENTER, &fc_found);

					if (!fc_found ||
						TimestampDifferenceExceeds(fc_hentry->last_start_time, fc_now,
												   wal_retrieve_retry_interval))
					{
						logicalrep_worker_launch(MyLogicalRepWorker->dbid,
												 MySubscription->oid,
												 MySubscription->name,
												 MyLogicalRepWorker->userid,
												 fc_rstate->relid);
						fc_hentry->last_start_time = fc_now;
					}
				}
			}
		}
	}

	if (fc_started_tx)
	{
		CommitTransactionCommand();
		pgstat_report_stat(true);
	}
}

/*
 * 处理正在同步的表可能的状态变化。
 */
void process_syncing_tables(XLogRecPtr fc_current_lsn)
{
	if (am_tablesync_worker())
		fc_process_syncing_tables_for_sync(fc_current_lsn);
	else
		fc_process_syncing_tables_for_apply(fc_current_lsn);
}

/*
 * 基于逻辑关系映射为COPY创建列的列表。
 */
static List * fc_make_copy_attnamelist(LogicalRepRelMapEntry *fc_rel)
{
	List	   *fc_attnamelist = NIL;
	int			fc_i;

	for (fc_i = 0; fc_i < fc_rel->remoterel.natts; fc_i++)
	{
		fc_attnamelist = lappend(fc_attnamelist,
							  makeString(fc_rel->remoterel.attnames[fc_i]));
	}


	return fc_attnamelist;
}

/*
 * COPY FROM的数据源回调，负责从远程
 * 连接读取并将数据传回我们的本地COPY。
 */
static int fc_copy_read_data(void *fc_outbuf, int fc_minread, int fc_maxread)
{
	int			fc_bytesread = 0;
	int			fc_avail;

	/* 如果有上一条读取的剩余数据，请使用它。 */
	fc_avail = copybuf->len - copybuf->cursor;
	if (fc_avail)
	{
		if (fc_avail > fc_maxread)
			fc_avail = fc_maxread;
		memcpy(fc_outbuf, &copybuf->data[copybuf->cursor], fc_avail);
		copybuf->cursor += fc_avail;
		fc_maxread -= fc_avail;
		fc_bytesread += fc_avail;
	}

	while (fc_maxread > 0 && fc_bytesread < fc_minread)
	{
		pgsocket	fc_fd = PGINVALID_SOCKET;
		int			fc_len;
		char	   *fc_buf = NULL;

		for (;;)
		{
			/* 尝试读取数据。 */
			fc_len = walrcv_receive(LogRepWorkerWalRcvConn, &fc_buf, &fc_fd);

			CHECK_FOR_INTERRUPTS();

			if (fc_len == 0)
				break;
			else if (fc_len < 0)
				return fc_bytesread;
			else
			{
				/* 处理数据 */
				copybuf->data = fc_buf;
				copybuf->len = fc_len;
				copybuf->cursor = 0;

				fc_avail = copybuf->len - copybuf->cursor;
				if (fc_avail > fc_maxread)
					fc_avail = fc_maxread;
				memcpy(fc_outbuf, &copybuf->data[copybuf->cursor], fc_avail);
				fc_outbuf = (void *) ((char *) fc_outbuf + fc_avail);
				copybuf->cursor += fc_avail;
				fc_maxread -= fc_avail;
				fc_bytesread += fc_avail;
			}

			if (fc_maxread <= 0 || fc_bytesread >= fc_minread)
				return fc_bytesread;
		}

		/*
		 * 等待更多数据或闩。
		 */
		(void) WaitLatchOrSocket(MyLatch,
								 WL_SOCKET_READABLE | WL_LATCH_SET |
								 WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
								 fc_fd, 1000L, WAIT_EVENT_LOGICAL_SYNC_DATA);

		ResetLatch(MyLatch);
	}

	return fc_bytesread;
}


/*
 * 以类似于在复制期间RELATION消息提供的信息的方式
 * 获取远程关系的信息。此函数还返回可在COPY命令中使用的
 * 关系资格。
 */
static void fc_fetch_remote_table_info(char *fc_nspname, char *fc_relname,
						LogicalRepRelation *fc_lrel, List **fc_qual)
{
	WalRcvExecResult *fc_res;
	StringInfoData fc_cmd;
	TupleTableSlot *fc_slot;
	Oid			fc_tableRow[] = {OIDOID, CHAROID, CHAROID};
	Oid			fc_attrRow[] = {INT2OID, TEXTOID, OIDOID, BOOLOID};
	Oid			fc_qualRow[] = {TEXTOID};
	bool		fc_isnull;
	int			fc_natt;
	ListCell   *fc_lc;
	Bitmapset  *fc_included_cols = NULL;

	fc_lrel->nspname = fc_nspname;
	fc_lrel->relname = fc_relname;

	/* 首先获取Oid和副本身份。 */
	initStringInfo(&fc_cmd);
	appendStringInfo(&fc_cmd, "SELECT c.oid, c.relreplident, c.relkind"
					 "  FROM pg_catalog.pg_class c"
					 "  INNER JOIN pg_catalog.pg_namespace n"
					 "        ON (c.relnamespace = n.oid)"
					 " WHERE n.nspname = %s"
					 "   AND c.relname = %s",
					 quote_literal_cstr(fc_nspname),
					 quote_literal_cstr(fc_relname));
	fc_res = walrcv_exec(LogRepWorkerWalRcvConn, fc_cmd.data,
					  lengthof(fc_tableRow), fc_tableRow);

	if (fc_res->status != WALRCV_OK_TUPLES)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_FAILURE),
				 errmsg("could not fetch table info for table \"%s.%s\" from publisher: %s",
						fc_nspname, fc_relname, fc_res->err)));

	fc_slot = MakeSingleTupleTableSlot(fc_res->tupledesc, &TTSOpsMinimalTuple);
	if (!tuplestore_gettupleslot(fc_res->tuplestore, true, false, fc_slot))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("table \"%s.%s\" not found on publisher",
						fc_nspname, fc_relname)));

	fc_lrel->remoteid = DatumGetObjectId(slot_getattr(fc_slot, 1, &fc_isnull));
	Assert(!fc_isnull);
	fc_lrel->replident = DatumGetChar(slot_getattr(fc_slot, 2, &fc_isnull));
	Assert(!fc_isnull);
	fc_lrel->relkind = DatumGetChar(slot_getattr(fc_slot, 3, &fc_isnull));
	Assert(!fc_isnull);

	ExecDropSingleTupleTableSlot(fc_slot);
	walrcv_clear_result(fc_res);


	/*
	 * 获取每个关系的列列表。
	 *
	 * 我们需要在获取列名和类型的信息之前完成此操作，
	 * 以便可以跳过那些不应被复制的列。
	 */
	if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
	{
		WalRcvExecResult *fc_pubres;
		TupleTableSlot *fc_tslot;
		Oid			fc_attrsRow[] = {INT2VECTOROID};
		StringInfoData fc_pub_names;
		initStringInfo(&fc_pub_names);
		foreach(fc_lc, MySubscription->publications)
		{
			if (foreach_current_index(fc_lc) > 0)
				appendStringInfo(&fc_pub_names, ", ");
			appendStringInfoString(&fc_pub_names, quote_literal_cstr(strVal(lfirst(fc_lc))));
		}

		/*
		 * 从所有发布中获取有关关系的列列表信息。
		 */
		resetStringInfo(&fc_cmd);
		appendStringInfo(&fc_cmd,
						 "SELECT DISTINCT"
						 "  (CASE WHEN (array_length(gpt.attrs, 1) = c.relnatts)"
						 "   THEN NULL ELSE gpt.attrs END)"
						 "  FROM pg_publication p,"
						 "  LATERAL pg_get_publication_tables(p.pubname) gpt,"
						 "  pg_class c"
						 " WHERE gpt.relid = %u AND c.oid = gpt.relid"
						 "   AND p.pubname IN ( %s )",
						 fc_lrel->remoteid,
						 fc_pub_names.data);

		fc_pubres = walrcv_exec(LogRepWorkerWalRcvConn, fc_cmd.data,
							 lengthof(fc_attrsRow), fc_attrsRow);

		if (fc_pubres->status != WALRCV_OK_TUPLES)
			ereport(ERROR,
					(errcode(ERRCODE_CONNECTION_FAILURE),
					 errmsg("could not fetch column list info for table \"%s.%s\" from publisher: %s",
							fc_nspname, fc_relname, fc_pubres->err)));

		/*
		 * 我们不支持在组合发布时同一表的列列表不同的情况。
		 * 见fetch_table_list顶部的注释。因此只应返回一行。
		 * 尽管我们在创建订阅时已经检查过这一点，但在同步工作线程
		 * 启动之前，我们仍然需要检查，以防列列表在创建订阅后
		 * 更改。
		 */
		if (tuplestore_tuple_count(fc_pubres->tuplestore) > 1)
			ereport(ERROR,
					errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					errmsg("cannot use different column lists for table \"%s.%s\" in different publications",
						   fc_nspname, fc_relname));

		/*
		 * 获取列列表并构建一个包含attnums的单一位图。
		 *
		 * 如果我们发现NULL值，则表示所有列都应被复制。
		 */
		fc_tslot = MakeSingleTupleTableSlot(fc_pubres->tupledesc, &TTSOpsMinimalTuple);
		if (tuplestore_gettupleslot(fc_pubres->tuplestore, true, false, fc_tslot))
		{
			Datum		fc_cfval = slot_getattr(fc_tslot, 1, &fc_isnull);

			if (!fc_isnull)
			{
				ArrayType  *fc_arr;
				int			fc_nelems;
				int16	   *fc_elems;

				fc_arr = DatumGetArrayTypeP(fc_cfval);
				fc_nelems = ARR_DIMS(fc_arr)[0];
				fc_elems = (int16 *) ARR_DATA_PTR(fc_arr);

				for (fc_natt = 0; fc_natt < fc_nelems; fc_natt++)
					fc_included_cols = bms_add_member(fc_included_cols, fc_elems[fc_natt]);
			}

			ExecClearTuple(fc_tslot);
		}
		ExecDropSingleTupleTableSlot(fc_tslot);

		walrcv_clear_result(fc_pubres);

		pfree(fc_pub_names.data);
	}

	/*
	 * 现在获取列名和类型。
	 */
	resetStringInfo(&fc_cmd);
	appendStringInfo(&fc_cmd,
					 "SELECT a.attnum,"
					 "       a.attname,"
					 "       a.atttypid,"
					 "       a.attnum = ANY(i.indkey)"
					 "  FROM pg_catalog.pg_attribute a"
					 "  LEFT JOIN pg_catalog.pg_index i"
					 "       ON (i.indexrelid = pg_get_replica_identity_index(%u))"
					 " WHERE a.attnum > 0::pg_catalog.int2"
					 "   AND NOT a.attisdropped %s"
					 "   AND a.attrelid = %u"
					 " ORDER BY a.attnum",
					 fc_lrel->remoteid,
					 (walrcv_server_version(LogRepWorkerWalRcvConn) >= 120000 ?
					  "AND a.attgenerated = ''" : ""),
					 fc_lrel->remoteid);
	fc_res = walrcv_exec(LogRepWorkerWalRcvConn, fc_cmd.data,
					  lengthof(fc_attrRow), fc_attrRow);

	if (fc_res->status != WALRCV_OK_TUPLES)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_FAILURE),
				 errmsg("could not fetch table info for table \"%s.%s\" from publisher: %s",
						fc_nspname, fc_relname, fc_res->err)));

	/* 我们不知道要来的行数，所以分配足够的空间。 */
	fc_lrel->attnames = palloc0(MaxTupleAttributeNumber * sizeof(char *));
	fc_lrel->atttyps = palloc0(MaxTupleAttributeNumber * sizeof(Oid));
	fc_lrel->attkeys = NULL;

	/*
	 * 将列存储为名称列表。若存在列列表，则忽略
	 * 不在列列表中的列。
	 */
	fc_natt = 0;
	fc_slot = MakeSingleTupleTableSlot(fc_res->tupledesc, &TTSOpsMinimalTuple);
	while (tuplestore_gettupleslot(fc_res->tuplestore, true, false, fc_slot))
	{
		char	   *fc_rel_colname;
		AttrNumber	fc_attnum;

		fc_attnum = DatumGetInt16(slot_getattr(fc_slot, 1, &fc_isnull));
		Assert(!fc_isnull);

		/* 如果列不在列列表中，则跳过它。 */
		if (fc_included_cols != NULL && !bms_is_member(fc_attnum, fc_included_cols))
		{
			ExecClearTuple(fc_slot);
			continue;
		}

		fc_rel_colname = TextDatumGetCString(slot_getattr(fc_slot, 2, &fc_isnull));
		Assert(!fc_isnull);

		fc_lrel->attnames[fc_natt] = fc_rel_colname;
		fc_lrel->atttyps[fc_natt] = DatumGetObjectId(slot_getattr(fc_slot, 3, &fc_isnull));
		Assert(!fc_isnull);

		if (DatumGetBool(slot_getattr(fc_slot, 4, &fc_isnull)))
			fc_lrel->attkeys = bms_add_member(fc_lrel->attkeys, fc_natt);

		/* 绝对不应发生。 */
		if (++fc_natt >= MaxTupleAttributeNumber)
			elog(ERROR, "too many columns in remote table \"%s.%s\"",
				 fc_nspname, fc_relname);

		ExecClearTuple(fc_slot);
	}
	ExecDropSingleTupleTableSlot(fc_slot);

	fc_lrel->natts = fc_natt;

	walrcv_clear_result(fc_res);

	/*
	 * 获取关系的行过滤表达式。DISTINCT 避免了同一表在多个出版物中出现相同
	 * 表达式，从而在最终表达式中被包含多次。
	 *
	 * 即使只匹配了其中一个出版物，我们也需要复制行，
	 * 以便稍后将所有条件与 OR 结合起来。
	 *
	 * 对于初始同步，在以下情况下可以忽略行过滤：
	 *
	 * 1) 其中一个已订阅的出版物未指定
	 * 任何行过滤
	 *
	 * 2) 其中一个已订阅的出版物将 puballtables 设置为 true
	 *
	 * 3) 其中一个已订阅的出版物声明为包含该关系的 TABLES IN SCHEMA
	 */
	if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
	{
		StringInfoData fc_pub_names;

		/* 构建出版名称列表。 */
		initStringInfo(&fc_pub_names);
		foreach(fc_lc, MySubscription->publications)
		{
			char	   *fc_pubname = strVal(lfirst(fc_lc));

			if (foreach_current_index(fc_lc) > 0)
				appendStringInfoString(&fc_pub_names, ", ");

			appendStringInfoString(&fc_pub_names, quote_literal_cstr(fc_pubname));
		}

		/* 检查行过滤。 */
		resetStringInfo(&fc_cmd);
		appendStringInfo(&fc_cmd,
						 "SELECT DISTINCT pg_get_expr(gpt.qual, gpt.relid)"
						 "  FROM pg_publication p,"
						 "  LATERAL pg_get_publication_tables(p.pubname) gpt"
						 " WHERE gpt.relid = %u"
						 "   AND p.pubname IN ( %s )",
						 fc_lrel->remoteid,
						 fc_pub_names.data);

		fc_res = walrcv_exec(LogRepWorkerWalRcvConn, fc_cmd.data, 1, fc_qualRow);

		if (fc_res->status != WALRCV_OK_TUPLES)
			ereport(ERROR,
					(errmsg("could not fetch table WHERE clause info for table \"%s.%s\" from publisher: %s",
							fc_nspname, fc_relname, fc_res->err)));

		/*
		 * 针对同一表的多个行过滤表达式将通过 COPY 使用 OR 进行组合。
		 * 如果该表的任何过滤表达式为 null，则表示整个表将被复制。在这种情况下，
		 * 完全没有必要构建统一的行过滤表达式。
		 */
		fc_slot = MakeSingleTupleTableSlot(fc_res->tupledesc, &TTSOpsMinimalTuple);
		while (tuplestore_gettupleslot(fc_res->tuplestore, true, false, fc_slot))
		{
			Datum		fc_rf = slot_getattr(fc_slot, 1, &fc_isnull);

			if (!fc_isnull)
				*fc_qual = lappend(*fc_qual, makeString(TextDatumGetCString(fc_rf)));
			else
			{
				/* 忽略过滤并根据需要进行清理。 */
				if (*fc_qual)
				{
					list_free_deep(*fc_qual);
					*fc_qual = NIL;
				}
				break;
			}

			ExecClearTuple(fc_slot);
		}
		ExecDropSingleTupleTableSlot(fc_slot);

		walrcv_clear_result(fc_res);
	}

	pfree(fc_cmd.data);
}

/*
 * 从发布者复制表的现有数据。
 *
 * 调用者负责锁定本地关系。
 */
static void fc_copy_table(Relation fc_rel)
{
	LogicalRepRelMapEntry *fc_relmapentry;
	LogicalRepRelation fc_lrel;
	List	   *fc_qual = NIL;
	WalRcvExecResult *fc_res;
	StringInfoData fc_cmd;
	CopyFromState fc_cstate;
	List	   *fc_attnamelist;
	ParseState *fc_pstate;

	/* 获取发布者关系信息。 */
	fc_fetch_remote_table_info(get_namespace_name(RelationGetNamespace(fc_rel)),
							RelationGetRelationName(fc_rel), &fc_lrel, &fc_qual);

	/* 将关系放入 relmap。 */
	logicalrep_relmap_update(&fc_lrel);

	/* 将发布者关系映射到本地关系。 */
	fc_relmapentry = logicalrep_rel_open(fc_lrel.remoteid, NoLock);
	Assert(fc_rel == fc_relmapentry->localrel);

	/* 在发布者上开始复制。 */
	initStringInfo(&fc_cmd);

	/* 没有行过滤的常规表 */
	if (fc_lrel.relkind == RELKIND_RELATION && fc_qual == NIL)
	{
		appendStringInfo(&fc_cmd, "COPY %s",
						 quote_qualified_identifier(fc_lrel.nspname, fc_lrel.relname));

		/* 如果表有列，则指定列 */
		if (fc_lrel.natts)
		{
			appendStringInfoString(&fc_cmd, " (");

			/*
			 * XXX 我们是否需要在所有情况下列出列？也许我们
			 * 正在复制所有列？
			 */
			for (int fc_i = 0; fc_i < fc_lrel.natts; fc_i++)
			{
				if (fc_i > 0)
					appendStringInfoString(&fc_cmd, ", ");

				appendStringInfoString(&fc_cmd, quote_identifier(fc_lrel.attnames[fc_i]));
			}

			appendStringInfoString(&fc_cmd, ")");
		}

		appendStringInfo(&fc_cmd, " TO STDOUT");
	}
	else
	{
		/*
		 * 对于非表和带有行过滤的表，我们需要执行 COPY
		 * (SELECT ...)，但我们不能仅仅执行 SELECT *，因为我们需要避免
		 * 复制生成的列。对于具有任何行过滤的表，构建带有 OR 连接的行过滤的
		 * SELECT 查询用于 COPY。
		 */
		appendStringInfoString(&fc_cmd, "COPY (SELECT ");
		for (int fc_i = 0; fc_i < fc_lrel.natts; fc_i++)
		{
			appendStringInfoString(&fc_cmd, quote_identifier(fc_lrel.attnames[fc_i]));
			if (fc_i < fc_lrel.natts - 1)
				appendStringInfoString(&fc_cmd, ", ");
		}

		appendStringInfoString(&fc_cmd, " FROM ");

		/*
		 * 对于常规表，确保我们不从继承命名表的子表复制数据，
		 * 因为这些将被单独复制。
		 */
		if (fc_lrel.relkind == RELKIND_RELATION)
			appendStringInfoString(&fc_cmd, "ONLY ");

		appendStringInfoString(&fc_cmd, quote_qualified_identifier(fc_lrel.nspname, fc_lrel.relname));
		/* OR 连接的过滤器列表 */
		if (fc_qual != NIL)
		{
			ListCell   *fc_lc;
			char	   *fc_q = strVal(linitial(fc_qual));

			appendStringInfo(&fc_cmd, " WHERE %s", fc_q);
			for_each_from(fc_lc, fc_qual, 1)
			{
				fc_q = strVal(lfirst(fc_lc));
				appendStringInfo(&fc_cmd, " OR %s", fc_q);
			}
			list_free_deep(fc_qual);
		}

		appendStringInfoString(&fc_cmd, ") TO STDOUT");
	}
	fc_res = walrcv_exec(LogRepWorkerWalRcvConn, fc_cmd.data, 0, NULL);
	pfree(fc_cmd.data);
	if (fc_res->status != WALRCV_OK_COPY_OUT)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_FAILURE),
				 errmsg("could not start initial contents copy for table \"%s.%s\": %s",
						fc_lrel.nspname, fc_lrel.relname, fc_res->err)));
	walrcv_clear_result(fc_res);

	copybuf = makeStringInfo();

	fc_pstate = make_parsestate(NULL);
	(void) addRangeTableEntryForRelation(fc_pstate, fc_rel, AccessShareLock,
										 NULL, false, false);

	fc_attnamelist = fc_make_copy_attnamelist(fc_relmapentry);
	fc_cstate = BeginCopyFrom(fc_pstate, fc_rel, NULL, NULL, false, fc_copy_read_data, fc_attnamelist, NIL);

	/* 执行复制 */
	(void) CopyFrom(fc_cstate);

	logicalrep_rel_close(fc_relmapentry, NoLock);
}

/*
 * 确定 tablesync 槽名称。
 *
 * 由于远程节点对槽名称长度的限制，名称不得超过 NAMEDATALEN - 1。我们附加
 * system_identifier 以避免与其他集群中的订阅发生槽名称冲突。
 * 根据当前方案 pg_%u_sync_%u_UINT64_FORMAT (3 + 10 + 6 + 10 + 20 + '\0')，槽名称的
 * 最大长度将为 50。
 *
 * 返回的槽名称存储在提供的缓冲区 (syncslotname) 中，大小由
 * 给定的大小决定。
 *
 * 注意：我们不使用订阅槽名称作为 tablesync 槽名称的一部分，
 * 因为我们负责清理这些槽，如果订阅槽名称发生变化，
 * 重新计算要清理的名称可能会变得不可能。
 */
void ReplicationSlotNameForTablesync(Oid fc_suboid, Oid fc_relid,
								char *fc_syncslotname, int fc_szslot)
{
	snprintf(fc_syncslotname, fc_szslot, "pg_%u_sync_%u_" UINT64_FORMAT, fc_suboid,
			 fc_relid, GetSystemIdentifier());
}

/*
 * 为 tablesync 形成原始名称。
 *
 * 在提供的缓冲区中返回名称。
 */
void ReplicationOriginNameForTablesync(Oid fc_suboid, Oid fc_relid,
								  char *fc_originname, int fc_szorgname)
{
	snprintf(fc_originname, fc_szorgname, "pg_%u_%u", fc_suboid, fc_relid);
}

/*
 * 开始在同步工作线程中同步表。
 *
 * 如果不需要对表进行同步，我们将退出工作线程，而不再
 * 进行进一步操作。
 *
 * 返回的槽名称是在当前内存上下文中分配的。
 */
char * LogicalRepSyncTableStart(XLogRecPtr *fc_origin_startpos)
{
	char	   *fc_slotname;
	char	   *fc_err;
	char		fc_relstate;
	XLogRecPtr	fc_relstate_lsn;
	Relation	fc_rel;
	AclResult	fc_aclresult;
	WalRcvExecResult *fc_res;
	char		fc_originname[NAMEDATALEN];
	RepOriginId fc_originid;

	/* 检查表同步的状态。 */
	StartTransactionCommand();
	fc_relstate = GetSubscriptionRelState(MyLogicalRepWorker->subid,
									   MyLogicalRepWorker->relid,
									   &fc_relstate_lsn);
	CommitTransactionCommand();

	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
	MyLogicalRepWorker->relstate = fc_relstate;
	MyLogicalRepWorker->relstate_lsn = fc_relstate_lsn;
	SpinLockRelease(&MyLogicalRepWorker->relmutex);

	/*
	 * 如果同步已经完成或不再必要，现在就退出，
	 * 因为我们已经更新了共享内存状态。
	 */
	switch (fc_relstate)
	{
		case SUBREL_STATE_SYNCDONE:
		case SUBREL_STATE_READY:
		case SUBREL_STATE_UNKNOWN:
			fc_finish_sync_worker();	/* 不返回 */
	}

	/* 计算 tablesync 槽的名称。 */
	fc_slotname = (char *) palloc(NAMEDATALEN);
	ReplicationSlotNameForTablesync(MySubscription->oid,
									MyLogicalRepWorker->relid,
									fc_slotname,
									NAMEDATALEN);

	/*
	 * 在这里，我们使用槽名称而不是订阅名称作为
	 * application_name，以便它不同于主要应用工作者，
	 * 这样同步复制可以区分它们。
	 */
	LogRepWorkerWalRcvConn =
		walrcv_connect(MySubscription->conninfo, true, fc_slotname, &fc_err);
	if (LogRepWorkerWalRcvConn == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_FAILURE),
				 errmsg("could not connect to the publisher: %s", fc_err)));

	Assert(MyLogicalRepWorker->relstate == SUBREL_STATE_INIT ||
		   MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC ||
		   MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY);

	/* 指定源跟踪记录名称。 */
	ReplicationOriginNameForTablesync(MySubscription->oid,
									  MyLogicalRepWorker->relid,
									  fc_originname,
									  sizeof(fc_originname));

	if (MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC)
	{
		/*
		 * 我们之前在完成复制之前出现错误，因此
		 * 复制槽可能存在。如果它已经存在，我们想要
		 * 删除槽并继续。
		 *
		 * XXX 我们也可以尝试删除槽，上次我们失败了，
		 * 但为此，我们可能需要清理复制状态，因为它可能
		 * 正在提取行的过程中。此外，如果发生网络
		 * 故障则不会成功，因此下次尝试似乎更有保障。
		 */
		ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, fc_slotname, true);
	}
	else if (MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY)
	{
		/*
		 * COPY 阶段之前已完成，但 tablesync 在
		 * 能够正常完成之前崩溃了。
		 */
		StartTransactionCommand();

		/*
		 * 源跟踪名称必须已经存在。它是在启动此
		 * tablesync 时首次创建的。
		 */
		fc_originid = replorigin_by_name(fc_originname, false);
		replorigin_session_setup(fc_originid);
		replorigin_session_origin = fc_originid;
		*fc_origin_startpos = replorigin_session_get_progress(false);

		CommitTransactionCommand();

		goto copy_table_done;
	}

	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
	MyLogicalRepWorker->relstate = SUBREL_STATE_DATASYNC;
	MyLogicalRepWorker->relstate_lsn = InvalidXLogRecPtr;
	SpinLockRelease(&MyLogicalRepWorker->relmutex);

	/* 更新状态并使其对其他人可见。 */
	StartTransactionCommand();
	UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
							   MyLogicalRepWorker->relid,
							   MyLogicalRepWorker->relstate,
							   MyLogicalRepWorker->relstate_lsn);
	CommitTransactionCommand();
	pgstat_report_stat(true);

	StartTransactionCommand();

	/*
	 * 在这里使用标准写锁定。可能最好在
	 * 同步期间禁止对表的访问。但我们不想阻塞
	 * 主要应用进程的工作，它在将远程关系 ID
	 * 映射到本地时必须以 RowExclusiveLock 打开关系。
	 */
	fc_rel = table_open(MyLogicalRepWorker->relid, RowExclusiveLock);

	/*
	 * 检查我们的表同步工作线程是否有插入目标表的
	 * 权限。
	 */
	fc_aclresult = pg_class_aclcheck(RelationGetRelid(fc_rel), GetUserId(),
								  ACL_INSERT);
	if (fc_aclresult != ACLCHECK_OK)
		aclcheck_error(fc_aclresult,
					   get_relkind_objtype(fc_rel->rd_rel->relkind),
					   RelationGetRelationName(fc_rel));

	/*
	 * COPY FROM 不遵循 RLS 策略。这对拥有 BYPASSRLS
	 * 权限的角色（或隐式拥有的超级用户）不是问题，
	 * 但其他角色不应能够规避 RLS。对这些角色
	 * 禁止在启用 RLS 的关系中进行逻辑复制。
	 */
	if (check_enable_rls(RelationGetRelid(fc_rel), InvalidOid, false) == RLS_ENABLED)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
						GetUserNameFromId(GetUserId(), true),
						RelationGetRelationName(fc_rel))));

	/*
	 * 在远程节点启动一个 REPEATABLE READ 模式的事务。
	 * 这确保我们创建的复制槽（见下文）和 COPY 之间
	 * 是一致的。
	 */
	fc_res = walrcv_exec(LogRepWorkerWalRcvConn,
					  "BEGIN READ ONLY ISOLATION LEVEL REPEATABLE READ",
					  0, NULL);
	if (fc_res->status != WALRCV_OK_COMMAND)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_FAILURE),
				 errmsg("table copy could not start transaction on publisher: %s",
						fc_res->err)));
	walrcv_clear_result(fc_res);

	/*
	 * 创建一个新的永久逻辑解码槽。此槽将在
	 * COPY 完成后的追赶阶段中使用，因此告知
	 * 它使用快照以使最终数据一致。
	 */
	walrcv_create_slot(LogRepWorkerWalRcvConn,
#ifdef FDD
					   fc_slotname, false /* 永久 */ , false /* 两阶段 */ , false /* 备份 */,
#else
					   fc_slotname, false /* 永久 */ , false /* 两阶段 */ ,
#endif
					   CRS_USE_SNAPSHOT, fc_origin_startpos);

	/*
	 * 设置复制源跟踪。在复制之前执行此操作的目的是避免
	 * 由于设置源跟踪时出现任何错误而再次进行复制。
	 */
	fc_originid = replorigin_by_name(fc_originname, true);
	if (!OidIsValid(fc_originid))
	{
		/*
		 * 来源跟踪不存在，因此现在创建它。
		 *
		 * 然后推进从 walrcv_create_slot 获得的 LSN。这是为了恢复而记录在 WAL 中。
		 * 锁是为了防止复制来源在推进时消失。
		 */
#ifdef FDDM
		fc_originid = fdd_replorigin_create_for_table_sync(fc_originname);
#else
		fc_originid = replorigin_create(fc_originname);
#endif

		LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
		replorigin_advance(fc_originid, *fc_origin_startpos, InvalidXLogRecPtr,
						   true /* 向后走 */ , true /* WAL 日志 */ );
		UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);

		replorigin_session_setup(fc_originid);
		replorigin_session_origin = fc_originid;
	}
	else
	{
		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_OBJECT),
				 errmsg("replication origin \"%s\" already exists",
						fc_originname)));
	}

	/* 现在执行初始数据复制 */
	PushActiveSnapshot(GetTransactionSnapshot());
	fc_copy_table(fc_rel);
	PopActiveSnapshot();

	fc_res = walrcv_exec(LogRepWorkerWalRcvConn, "COMMIT", 0, NULL);
	if (fc_res->status != WALRCV_OK_COMMAND)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_FAILURE),
				 errmsg("table copy could not finish transaction on publisher: %s",
						fc_res->err)));
	walrcv_clear_result(fc_res);

	table_close(fc_rel, NoLock);

	/* 使复制可见。 */
	CommandCounterIncrement();

	/*
	 * 更新已保存的状态以指示 COPY 阶段已完成；使其对其他人可见。
	 */
	UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
							   MyLogicalRepWorker->relid,
							   SUBREL_STATE_FINISHEDCOPY,
							   MyLogicalRepWorker->relstate_lsn);

	CommitTransactionCommand();

copy_table_done:

	elog(DEBUG1,
		 "LogicalRepSyncTableStart: '%s' origin_startpos lsn %X/%X",
		 fc_originname, LSN_FORMAT_ARGS(*fc_origin_startpos));

	/*
	 * 我们已完成初始数据同步，更新状态。
	 */
	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
	MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCWAIT;
	MyLogicalRepWorker->relstate_lsn = *fc_origin_startpos;
	SpinLockRelease(&MyLogicalRepWorker->relmutex);

	/*
	 * 最后，等待主应用工作线程告诉我们赶上来，然后返回以让 LogicalRepApplyLoop 做这件事。
	 */
	fc_wait_for_worker_state_change(SUBREL_STATE_CATCHUP);
	return fc_slotname;
}

/*
 * 常用代码将最新的同步状态信息提取到静态列表中。
 *
 * 如果订阅有 1 或多个表，则返回 true，否则返回 false。
 *
 * 注意：如果此函数启动了事务（通过参数指示），则提交它的责任在于调用者。
 */
static bool fc_FetchTableStates(bool *fc_started_tx)
{
	static bool fc_has_subrels = false;

	*fc_started_tx = false;

	if (table_states_validity != SYNC_TABLE_STATE_VALID)
	{
		MemoryContext fc_oldctx;
		List	   *fc_rstates;
		ListCell   *fc_lc;
		SubscriptionRelState *fc_rstate;

		table_states_validity = SYNC_TABLE_STATE_REBUILD_STARTED;

		/* 清理旧列表。 */
		list_free_deep(table_states_not_ready);
		table_states_not_ready = NIL;

		if (!IsTransactionState())
		{
			StartTransactionCommand();
			*fc_started_tx = true;
		}

		/* 获取所有未准备好的表。 */
		fc_rstates = GetSubscriptionNotReadyRelations(MySubscription->oid);

		/* 在永久内存上下文中分配跟踪信息。 */
		fc_oldctx = MemoryContextSwitchTo(CacheMemoryContext);
		foreach(fc_lc, fc_rstates)
		{
			fc_rstate = palloc(sizeof(SubscriptionRelState));
			memcpy(fc_rstate, lfirst(fc_lc), sizeof(SubscriptionRelState));
			table_states_not_ready = lappend(table_states_not_ready, fc_rstate);
		}
		MemoryContextSwitchTo(fc_oldctx);

		/*
		 * 订阅是否有表？
		 *
		 * 如果找到未准备好的关系，则我们知道它有表。但如果 table_state_not_ready 为空，我们仍然需要再次检查以查看是否有 0 个表。
		 */
		fc_has_subrels = (list_length(table_states_not_ready) > 0) ||
			HasSubscriptionRelations(MySubscription->oid);

		/*
		 * 如果自从我们进入此例程以来，订阅关系缓存已失效，我们仍然使用并返回我们刚刚完成构建的关系，以避免无尽循环，但我们将表状态标记为已过时，以便在下次访问时重新构建它。否则，我们将表状态标记为有效。
		 */
		if (table_states_validity == SYNC_TABLE_STATE_REBUILD_STARTED)
			table_states_validity = SYNC_TABLE_STATE_VALID;
	}

	return fc_has_subrels;
}

/*
 * 如果订阅没有表，则返回 false。
 *
 * 否则，所有表同步是否已准备好？
 *
 * 注意：此函数不适合从应用或表同步工作线程外部调用，因为 MySubscription 需要已初始化。
 */
bool AllTablesyncsReady(void)
{
	bool		fc_started_tx = false;
	bool		fc_has_subrels = false;

	/* 我们在这里需要订阅表的最新同步状态信息。 */
	fc_has_subrels = fc_FetchTableStates(&fc_started_tx);

	if (fc_started_tx)
	{
		CommitTransactionCommand();
		pgstat_report_stat(true);
	}

	/*
	 * 当订阅中没有表或并非所有表处于准备状态时返回 false；否则返回 true。
	 */
	return fc_has_subrels && list_length(table_states_not_ready) == 0;
}

/*
 * 更新 pg_subscription 中指定订阅的 two_phase 状态。
 */
void UpdateTwoPhaseState(Oid fc_suboid, char fc_new_state)
{
	Relation	fc_rel;
	HeapTuple	fc_tup;
	bool		fc_nulls[Natts_pg_subscription];
	bool		fc_replaces[Natts_pg_subscription];
	Datum		fc_values[Natts_pg_subscription];

	Assert(fc_new_state == LOGICALREP_TWOPHASE_STATE_DISABLED ||
		   fc_new_state == LOGICALREP_TWOPHASE_STATE_PENDING ||
		   fc_new_state == LOGICALREP_TWOPHASE_STATE_ENABLED);

	fc_rel = table_open(SubscriptionRelationId, RowExclusiveLock);
	fc_tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(fc_suboid));
	if (!HeapTupleIsValid(fc_tup))
		elog(ERROR,
			 "cache lookup failed for subscription oid %u",
			 fc_suboid);

	/* 形成一个新元组。 */
	memset(fc_values, 0, sizeof(fc_values));
	memset(fc_nulls, false, sizeof(fc_nulls));
	memset(fc_replaces, false, sizeof(fc_replaces));

	/* 并更新/设置 two_phase 状态 */
	fc_values[Anum_pg_subscription_subtwophasestate - 1] = CharGetDatum(fc_new_state);
	fc_replaces[Anum_pg_subscription_subtwophasestate - 1] = true;

	fc_tup = heap_modify_tuple(fc_tup, RelationGetDescr(fc_rel),
							fc_values, fc_nulls, fc_replaces);
	CatalogTupleUpdate(fc_rel, &fc_tup->t_self, fc_tup);

	heap_freetuple(fc_tup);
	table_close(fc_rel, RowExclusiveLock);
}
